package consumer

import (
	"encoding/json"
	"new_for_deepseek_rabbitmq/logger"
	"new_for_deepseek_rabbitmq/model"
	syncModel "new_for_deepseek_rabbitmq/model/sync"
	rabbitmqProject "new_for_deepseek_rabbitmq/project/rabbitmq"
	"new_for_deepseek_rabbitmq/project/yongyou/mk"
	"sync"
	"time"

	"github.com/streadway/amqp"
	"gorm.io/gorm"
)

type SyncConsumerStruct struct {
	sync.Mutex       // 添加互斥锁
	mkClient         *mk.Mk
	MkArticleRequest mk.MkArticleRequest
	logId            int
}

func SyncConsumer() *SyncConsumerStruct {
	return &SyncConsumerStruct{
		mkClient: &mk.Mk{},
		logId:    0,
		MkArticleRequest: mk.MkArticleRequest{
			AuthorNickName:      "AI智能体",
			CoverImg:            "group1/M00/02/CF/CgoRDGaXa4GANNDnAAEHx9EdlUU393.jpg",
			Categories:          []string{"c263efbcc73947339beb94afe5bb54ae"},
			RecommendList:       []string{},
			VisibleRange:        0,
			WechatSubscribe:     0,
			RecommendUserRegist: 1,
			Fields:              []string{""},
			Industries:          []string{""},
			Labels:              []string{},
			OpenForTenant:       true,
			Status:              "1",
			Files:               []string{},
			LinkUrl:             "",
			Type:                "0",
			PageStatus:          false,
			PageHeadline:        "",
		},
	}
}

// 入口方法
func (s *SyncConsumerStruct) Start() (err error) {
	ch, err := rabbitmqProject.GetChannel()
	if err != nil {
		logger.Error("获取RabbitMQ通道失败: %s", err.Error())
		return
	}
	defer rabbitmqProject.ReleaseChannel(ch)

	// 使用工具函数声明交换机和队列
	if err = rabbitmqProject.DeclareExchangeAndQueue(ch, rabbitmqProject.SyncConfig); err != nil {
		logger.Error("初始化RabbitMQ失败: %s", err.Error())
		return
	}

	// 消费消息
	msgs, err := ch.Consume(
		rabbitmqProject.SyncConfig.QueueName,
		"",
		false,
		false,
		false,
		false,
		nil,
	)
	if err != nil {
		logger.Error("设置消费者失败: %s", err.Error())
		return
	}

	// 循环消费消息
	for msg := range msgs {
		// 解析消息，获取 logId
		var syncMessage rabbitmqProject.SyncNewsMessage
		if err := json.Unmarshal(msg.Body, &syncMessage); err != nil {
			logger.Error("解析消息失败: %s", err.Error())
			msg.Nack(false, false)
			continue
		}

		// 同步处理消息
		s.logId = syncMessage.LogId
		if err := s.syncNews(msg); err != nil {
			msg.Nack(false, false)
		} else {
			msg.Ack(false)
		}
	}

	return nil
}

// 同步消息
func (s *SyncConsumerStruct) syncNews(msg amqp.Delivery) (err error) {
	var syncMessage rabbitmqProject.SyncNewsMessage
	if err := json.Unmarshal(msg.Body, &syncMessage); err != nil {
		logger.Error("Failed to unmarshal message body: %s, raw message: %s", err.Error(), string(msg.Body))
		s.incrementErrorNum(err.Error())
		return err
	}

	db := model.GetInstance().DB

	//查询所有新闻
	var news []model.NewsModel
	db.Model(&model.NewsModel{}).
		Where("id in ?", syncMessage.Ids).
		Where("sync_status = 0").
		Find(&news)

	// 如果新闻数量为0，则直接返回
	if len(news) == 0 {
		return nil
	}

	for _, n := range news {
		// 转换为mk文章请求
		s.MkArticleRequest.Title = n.Title
		s.MkArticleRequest.Content = n.Text
		s.MkArticleRequest.KeywordIndex = n.SeoKeyword
		s.MkArticleRequest.PageDescribe = n.SeoDesc
		s.MkArticleRequest.AuthorNickName = "AI智能体"

		// 发布文章
		mkId, err := s.mkClient.PublishArticle(s.MkArticleRequest)
		if err != nil {
			s.incrementErrorNum("发布文章失败:" + err.Error())
			logger.Error("发布文章失败: %s", err.Error())
			continue
		}

		// 更新状态
		db.Model(&model.NewsModel{}).
			Where("id = ?", n.Id).
			Updates(map[string]interface{}{
				"sync_status": 1,
				"sync_at":     time.Now(),
				"mk_id":       mkId,
			})

		//判断是最后一条 回到初始状态
		startId := n.Id + 1
		var newsModel model.NewsModel
		model.GetInstance().DB.Model(&model.NewsModel{}).Where("id = ?", n.Id).Order("id DESC").First(&newsModel)
		if startId > newsModel.Id {
			startId = 0
		}

		model.GetInstance().DB.Model(&syncModel.SyncConfigModel{}).
			Where("id = ?", 1).Update("start_id", startId)
	}

	s.incrementSuccessNum()
	return nil
}

// 修改 incrementErrorNum 方法
func (s *SyncConsumerStruct) incrementErrorNum(errorMsg string) {
	db := model.GetInstance().DB
	if err := db.Model(&syncModel.SyncLogsModel{}).Where("id = ?", s.logId).UpdateColumn("error_num", gorm.Expr("error_num + 1")).Error; err != nil {
		logger.Error("增加错误数量失败: %s", err.Error())
	}
	//增加错误信息
	if err := db.Model(&syncModel.SyncLogsModel{}).Create(map[string]interface{}{
		"sync_logs_id": s.logId,
		"error_msg":    errorMsg,
		"created_at":   time.Now(),
		"updated_at":   time.Now(),
	}).Error; err != nil {
		logger.Error("记录错误信息失败: %s", err.Error())
	}
}

// 修改 incrementSuccessNum 方法
func (s *SyncConsumerStruct) incrementSuccessNum() {
	db := model.GetInstance().DB
	if err := db.Model(&syncModel.SyncLogsModel{}).Where("id = ?", s.logId).UpdateColumn("success_num", gorm.Expr("success_num + 1")).Error; err != nil {
		logger.Error("增加成功数量失败: %s", err.Error())
	}
}
